package _01_quick_start._8_2_6_materialized_values

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream._
import akka.stream.scaladsl._

import scala.concurrent.Future

object Main extends App {

  implicit val system       = ActorSystem("QuickStart")
  implicit val materializer = ActorMaterializer()
  implicit val ec           = system.dispatcher

  final case class Author(handle: String)

  final case class Hashtag(name: String)

  final case class Tweet(author: Author, timestamp: Long, body: String) {
    def hashtags: Set[Hashtag] =
      body.split(" ").collect { case t if t.startsWith("#") => Hashtag(t) }.toSet
  }

  val akkaTag = Hashtag("#akka")

  /*
   * 非结构化`streaming`处理
   */

  // source
  val tweets: Source[Tweet, NotUsed] = Source(
    Set(
      Tweet(Author("Martin"), System.currentTimeMillis(), "Ordersky #akka creator"),
      Tweet(Author("Barudisu"), System.currentTimeMillis(), "#akka leaner"),
      Tweet(Author("Petter"), System.currentTimeMillis(), "Mahina #scala practise")
    ))

  val count: Flow[Tweet, Int, NotUsed] = Flow[Tweet].map(_ => 1)

  val sumSink: Sink[Int, Future[Int]] = Sink.fold[Int, Int](0)(_ + _)

  val counterGraph: RunnableGraph[Future[Int]] =
    tweets
    .via(count)
    .toMat(sumSink)(Keep.right)

  // 物化`Materialized`，就是将各种`source`，`sink`，`flow`组合构成一个graph，并最终执行
  val sum: Future[Int] = counterGraph.run()

  sum.foreach(c => println(s"Total tweets processed: $c"))
  sum.onComplete(_ => system.terminate())

  //  val sumSink = Sink.fold[Int, Int](0)(_ + _)
  //  val counterRunnableGraph: RunnableGraph[Future[Int]] =
  //    tweetsInMinuteFromNow
  //    .filter(_.hashtags contains akkaTag)
  //    .map(t => 1)
  //    .toMat(sumSink)(Keep.right)
  //
  //  // materialize the stream once in the morning
  //  val morningTweetsCount: Future[Int] = counterRunnableGraph.run()
  //  // and once in the evening, reusing the flow
  //  val eveningTweetsCount: Future[Int] = counterRunnableGraph.run()
  //
  //  val sum: Future[Int] = tweets.map(t => 1).runWith(sumSink)

}
